package com.sun.electric.compression.api
import java.io.OutputStream
import java.nio.ByteBuffer
import java.util.concurrent.Executors
import java.util.concurrent.ExecutorService
import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.Future
import java.util.Queue
import java.util.concurrent.LinkedBlockingQueue
import com.sun.electric.compression.bzip2.BZip2Exception
import com.sun.electric.compression.bzip2.BZip2OutputStream
import java.util.concurrent.TimeUnit
import com.sun.electric.compression.bzip2.BZip2Constants
import java.util.LinkedList
import java.util.concurrent.BlockingQueue

/**
 * @see BZip2OutputStream
 * @author Daniel.Schmidt
 */
class BZIP2OutputStreamScala(outputStream: OutputStream, blockSizeFactor: Int) extends BZip2OutputStream(outputStream, blockSizeFactor) {

  /* The threadpool with nThreads threads waiting to execute tasks */
  var threadpool: ExecutorService = null
  /* All tasks submitted to the threadpool are queued */
  var writingQueue: LinkedList[Future[BZIP2BlockCompressorScala]] = null
  /* Amount of tasks submitted */
  var amountOfTasks = 0
  /* The current compressor whose data shall be written to the file at next */
  var currentCompressor : Future[BZIP2BlockCompressorScala] = null
  /* Amount of threads */
  var nThreads = 0
  
  /**
   * C'tor
   * Initializes the threadpool, the writing queue and the BlockCompressor for the
   * first block of data.
   */
  def this(outputStream: OutputStream, blockSizeFactor: Int, nThreads: Int) = {
    this(outputStream, blockSizeFactor)
    if(nThreads > 1)
    	//Initialize the thread pool with n - 1 thread
        //Notice: the main thread does work himself!
    	threadpool = Executors.newFixedThreadPool(nThreads - 1)
    this.nThreads = nThreads
    writingQueue = new LinkedList[Future[BZIP2BlockCompressorScala]]()
    
    initialiseNextBlock()
  }

  /**
   * Initializes the next BlockCompressor containing the current block of data.
   */
  override def initialiseNextBlock() = {

    this.blockCompressor = new BZIP2BlockCompressorScala(
      this.bitOutputStream,
      this.streamBlockSize)
    amountOfTasks = amountOfTasks + 1
  }

  /**
   * If the current compressor is full the following alternatives are possible:
   * 
   * If there are "too much" elements waiting in the writing queue:
   * 	- Check if the current compressor has finished
   * 		-> If yes, write the compressed data to the file
   * 		-> If no, compress a block yourself and check later again
   * If the writing queue has an acceptable size, submit the procession of 
   * the current block and execute it in parallel with help of the thread pool.
   * The returned future indicates the completion of the task and is stored
   * in the writing queue.
   * 
   * Notice: This handling is needed to prevent a java out of memory exception.
   * This exception appears when all compressors are created initially. Now
   * the main thread has a threshold that determines the number of compressors
   * in the queue. If the threshold is exceeded, it tries to write finished data at first.
   */
  override def closeBlock() = {

    //Set the current compressor if it is still null. Therefore the writing queue
    //has to contain an element. 
    if (currentCompressor == null && !writingQueue.isEmpty()) {
      currentCompressor = writingQueue.poll();
    }

    var closed = false
    do {
      if (nThreads == 1 || writingQueue.size() > ((nThreads - 1) * 2)) {
        //Only check if current compressor is done, otherwise compress yourself during 
        //the meantime
        if (currentCompressor != null && currentCompressor.isDone()) {
          writeToStream()
        } else {
          val comp = this.blockCompressor.asInstanceOf[BZIP2BlockCompressorScala].call()
          writingQueue.add(new FutureOfCompressor(comp))
          closed = true
        }
      } else {
        val f: Future[BZIP2BlockCompressorScala] = threadpool.submit(this.blockCompressor.asInstanceOf[BZIP2BlockCompressorScala])
        writingQueue.add(f);
        closed = true
      }
    } while (!closed)
  }

  /**
   * After all blocks are compressed in parallel, the results can be written out to the 
   * file. Finally the thread pool can be shut down.
   */
  override def finish() = {

    if (!this.streamFinished) {
      this.streamFinished = true;
      try {

        closeBlock();
        while(amountOfTasks > 0) {
          if (currentCompressor == null) {
        	  currentCompressor = writingQueue.poll();
          }
          writeToStream();
        }
        
        this.bitOutputStream.writeBits(24, BZip2Constants.STREAM_END_MARKER_1);
        this.bitOutputStream.writeBits(24, BZip2Constants.STREAM_END_MARKER_2);
        this.bitOutputStream.writeInteger(this.streamCRC);
        this.bitOutputStream.flush();
        this.outputStream.flush();
      } finally {
        if(nThreads > 1)
        	threadpool.shutdownNow();
        this.blockCompressor = null;
      }
    }
  }


  /**
   * Writes the information(header, compressed data, CRC) of the finished compressor
   * to the file. Due to the queue it is ensured that the order of the compressed  
   * block data is maintained and written out correctly.
   */
  def writeToStream() {
  
    val compressor: BZIP2BlockCompressorScala = currentCompressor.get()

    amountOfTasks = amountOfTasks - 1

    // Write out the block header
    this.bitOutputStream.writeBits(24, BZip2Constants.BLOCK_HEADER_MARKER_1)
    this.bitOutputStream.writeBits(24, BZip2Constants.BLOCK_HEADER_MARKER_2)
    this.bitOutputStream.writeInteger(compressor.getHeaderCRC())
    this.bitOutputStream.writeBoolean(false); // Randomised block flag. We never create randomised blocks
    this.bitOutputStream.writeBits(24, compressor.getBwtStartPointer())

    // Write out the symbol map
    compressor.writeSymbolMap()

    compressor.getEncoder().writeEncoding()

    val blockCRC: Int = compressor.getCRC();
    this.streamCRC = ((this.streamCRC << 1) | (this.streamCRC >>> 31)) ^ blockCRC

    // Set new compressor
    currentCompressor = writingQueue.poll();
  }
  
  /**
   * Holds an BZIP2BlockCompressorScala and implements the future interface.
   * This class is needed for the main thread to process a block itself and
   * store the result as future in the general writing queue.
   */
  class FutureOfCompressor(compressor : BZIP2BlockCompressorScala) extends Future[BZIP2BlockCompressorScala] {
    
    override def cancel(mayInterruptIfRunning : Boolean) = false
    override def get() = compressor 
    override def get(timeout : Long, unit : TimeUnit) = compressor
    override def isCancelled() = false
    override def isDone() = true
  }
}

